
//Copyright 1997-2009 Syrinx Development, Inc.
//This file is part of the Syrinx Web Application Framework (SWAF).
// == BEGIN LICENSE ==
//
// Licensed under the terms of any of the following licenses at your
// choice:
//
//  - GNU General Public License Version 3 or later (the "GPL")
//    http://www.gnu.org/licenses/gpl.html
//
//  - GNU Lesser General Public License Version 3 or later (the "LGPL")
//    http://www.gnu.org/licenses/lgpl.html
//
//  - Mozilla Public License Version 1.1 or later (the "MPL")
//    http://www.mozilla.org/MPL/MPL-1.1.html
//
// == END LICENSE ==
using System;
using System.Collections;
using System.Diagnostics;
using System.Threading;

using Swaf.BizRule;
using Swaf.Call;
using Swaf.Container;
using Swaf.Crypto;
using Swaf.Logging;

namespace Swaf.Messaging
{
	/// <summary>
	/// Connects to a messaging system to listen for messages based on a configuration given to it
	/// during construction of the object.
	/// </summary>
	/// <remarks>Each instance of this class creates its own thread that is started when
	/// MessagingListener.start is called.  This thread recieves messages from the messaging system
	/// and calls each of its observers with the message, which can be formatted from the base message through
	/// the formatter string passed in to calls of MessagingListener.add.
	/// </remarks>
	public class MessagingListener : IMessagingListener
	{
		protected IMessagingSystem m_msgSys;
		protected string m_queueName;
		protected string m_connection;
		protected string m_processingFailureQueue;
		protected string m_historyQueueName;
		protected IList m_observers;
		protected ObserverInfo m_defaultObserver;
		protected IMessagingConnectionInfo m_connectionInfo;

		public MessagingListener(IMessagingSystem msgSys, string queueName, string connection, string processingFailureQueue, string historyQueue)
		{
			m_msgSys = msgSys;
			m_queueName = queueName;
			m_historyQueueName=historyQueue;
			m_connection = connection;
			m_processingFailureQueue = processingFailureQueue;
			m_observers = new ArrayList();
			m_connectionInfo = (QueueConnectionInfo) Application.currentApp.appData.get("Messaging." + connection);
			Debug.Assert(m_msgSys != null);
		}
		#region IMessagingListener Members

		public void addObserver(Swaf.Call.ICall callback, string formatter, bool providesReplyContent, string matchPattern, string encryptResults, bool isDefault)
		{
			ObserverInfo info = new ObserverInfo(callback, formatter, providesReplyContent, matchPattern,encryptResults);
			m_observers.Add(info);
			if (isDefault)
				m_defaultObserver = info;
		}

		public void removeObserver(Swaf.Call.ICall callback)
		{
			// TODO:  Add MessagingListener.removeObserver implementation
		}


		public bool processSingleMessage()
		{
			try
			{
				int timeoutInterval=1;
				string msgTimeOut = Application.currentApp.resMgr.resolve("$MessageTimeoutInterval$");
				if (! msgTimeOut.Equals("$MessageTimeoutInterval$"))
				{
					try
					{
						timeoutInterval = int.Parse(msgTimeOut);
					}
					catch(Exception)
					{}
				}
				m_connectionInfo.QueueName=m_queueName;
				m_msgSys.beginTrans(m_connectionInfo);
				IMessage msg=null;
				try
				{
					msg = m_msgSys.receive(m_connectionInfo, timeoutInterval);
				}
				catch(Exception em)
				{
					System.Diagnostics.Debug.WriteLine(em.ToString());
				}
				//IMessage msg = m_msgSys.receive(m_queueName,m_connection, 1);
				if (msg != null)
				{
					//Save the message to history.
					if (! m_historyQueueName.Equals(string.Empty))
					{
						try
						{
							m_connectionInfo.QueueName=m_historyQueueName;
							m_msgSys.send(msg, m_connectionInfo);
						}
						catch(Exception hEx)
						{
							System.Diagnostics.Debug.WriteLine(string.Format("Exception writing to history queue {0}. {1}",m_historyQueueName,hEx.ToString()));
						}
					}
					doMessageProcessingWork(msg);
					//Application.currentApp.ruleMgr.execute("messagingListenerProcessMessage", this, "messageProcessingComplete", new MessagingListenerMessageInfo(this, msg));
					return true;
				}
				else
				{
					try
					{
						m_msgSys.rollback(m_connectionInfo);
					}
					catch(Exception e)
					{
						System.Diagnostics.Debug.WriteLine(e.ToString());
					}
				}
			}
			catch(Exception e)
			{
				//Intentionally left blank.  For now, we assume all exceptions are about timeouts, and not other failures.
				//This is obviously flawed, but the MessagingException hiarchy needs to updated first.
				System.Diagnostics.Debug.WriteLine(e.ToString());
			}
			return false;
		}

		public void messageProcessingComplete(IResults rc)
		{
			if(rc != null);
		}

		internal void doMessageProcessingWork(IMessage msg)
		{
			//Call each observer with the message to do the real work
			BasicResults rc = new BasicResults();
			object callRc = null;
			string msgString = msg.ToString();
			bool wasRuleCall=false;
			bool observerRequiresEncryption=false;
			bool wasProcessed=false;
			bool sendToPoison=false;
            string encrypt="none";
			Exception processingException=null;
			try
			{
				//Messaging Rule Calls are a special instance.  If we see that this is one of those messages,
				//do not call the observers, just handle that message.
				if (msgString.StartsWith(MessagingRuleMgrCall.RuleMessageIdentifier))
				{
					wasRuleCall=true;
					IResults irc = null;
					irc = Application.currentApp.ruleMgr.execute(msgString);
					rc = new BasicResults(irc.CallResults,irc.ExceptionInfo);
					encrypt=msg.EncryptedInfo();
				}
				else
				{
					foreach(ObserverInfo info in m_observers)
					{
						try
						{
							//If more than 1 observer defines different encryption mechanisms, we have a problem!  
							//Last in wins!
							encrypt=info.EncryptResults.Equals("none") ? encrypt : info.EncryptResults;
							bool callObserver = true;
							if(info.MatchPattern != null)
								callObserver = Application.currentApp.resMgr.matches(info.MatchPattern, msgString).Success;
							//callRc = info.Observer.call(msgString);
							if(callObserver)
							{
								callRc = tryToCallObserver(info, msg);
							}
							if(info.ProvidesResult)
								rc.add(callObserver?callRc:null, null);
							wasProcessed = wasProcessed || callObserver;
						}
						catch(Exception e)
						{
							processingException=e;
							if(info.ProvidesResult)
								rc.add(null, e);
							sendToPoison=true;
						}
					}
					if (! wasProcessed)
					{
						if (m_defaultObserver != null)
						{
							try
							{
								callRc = tryToCallObserver(m_defaultObserver, msg);
								if(m_defaultObserver.ProvidesResult)
									rc.add(callRc, null);
							}
							catch(Exception e)
							{
								processingException=e;
								if(m_defaultObserver.ProvidesResult)
									rc.add(null, e);
								sendToPoison=true;
							}
						}
						else
						{
							sendToPoison=true;
						}
					}
					if (sendToPoison)
					{
						string messageId=Guid.NewGuid().ToString();
						string messageDest = msg.MessageSource;
						string failedMessage=string.Format("<FailedMessage messageID='{0}' messageSource='{1}'>{2}</FailedMessage>",messageId,messageDest, msg.MessageText);
						Application.currentApp.Log.println(LoggingLevels.Critical,"Exception processing message {0} from {1}. {2}", messageId,messageDest,processingException.ToString());
						m_msgSys.send(failedMessage,"none", m_processingFailureQueue, m_connection, "Text", new FlexiMap());
					}
				}
				//If the message needs a reply, build the reply based on the results of calling the observers.
				if(msg.HasReply())
				{
					string replyBody = "";
					if(rc.numberOfResults == 1)
					{
						replyBody = rc.getXml();
						/*
						if(rc.CallResults != null)
							replyBody = Serialization.serialize(rc.CallResults).SerialObject;
						else if(rc.ExceptionInfo != null)
							replyBody = Serialization.serialize(rc.ExceptionInfo).SerialObject;
						*/
					}
					else if(rc.numberOfResults > 1)
						replyBody = rc.getXml();
					//If the incoming call was a rule call and the message was originally encrypted,
					//encrypt the results.  If an observer handled the call, it will control the 
					//encryption.
					if ((wasRuleCall && !encrypt.Equals("none") ) )
						replyBody=MessagingSystemBase.MessageEncryptedIdentifier + encrypt + "]" + ((ICryptoManager)Application.currentApp.appData["sysObjs-Crypto"]).encrypt(encrypt,replyBody);
					//Console.WriteLine("Listener about to send reply");
					IMessagingConnectionInfo connection = msg.GetConnectionInfo();
					bool isSameConnection=(connection.QueueManager.Equals(m_connectionInfo.QueueManager) && connection.HostName.Equals(m_connectionInfo.HostName));
					if (! isSameConnection)
					{
						m_msgSys.beginTrans(connection);
					}
					m_msgSys.reply(replyBody, msg.GetDestination(), msg);
					//Console.WriteLine("Listener done sending reply");

					if (!isSameConnection)
					{
						try
						{
							m_msgSys.commit(connection);
						}
						catch(Exception ignore)
						{
							System.Diagnostics.Debug.WriteLine(ignore.ToString());
							//Do nothing.
						}
					}

				}
				m_msgSys.commit(m_connectionInfo);
			}
			catch(Exception e)
			{
				try
				{
					Application.currentApp.Log.println("Unable to process message in MessageListener for queue", msg.GetConnectionInfo().QueueName);
				}
				catch (Exception)
				{
					// ignore because we blew up logging the exception.
				}

				m_msgSys.rollback(m_connectionInfo);
			}
		}
		/// <summary>
		/// Helper class to try and call an observer
		/// </summary>
		/// <param name="msg"></param>
		/// <returns></returns>
		private object tryToCallObserver(ObserverInfo info,IMessage msg)
		{
			object callRc=null;
			string msgString = msg.ToString();

			if (! info.MessageStringOnly)
			{
				try
				{
					callRc = info.Observer.call(msg);
				}
				catch(Exception)
				{
					callRc = info.Observer.call(msgString);
					info.MessageStringOnly=true;
				}
			}
			else
			{
				callRc = info.Observer.call(msgString);
			}
			//Test for Rule Call type so that we can check for exceptions.
			IResults result = callRc as IResults;
			if (result != null)
			{
				if (result.ExceptionInfo != null)
					throw result.ExceptionInfo;
			}
			return callRc;
		}

		#endregion

		/// <summary>
		/// Maintains information about an observer to a MessagingListener instance that is used to
		/// do callbacks when messages arrive.
		/// </summary>
		protected class ObserverInfo
		{
			public ICall Observer;
			/// <summary>
			/// A class factory pattern to create a formatter that will convert the received message content
			/// into a suitable type for the observer when being called back.
			/// </summary>
			public string Formatter;
			/// <summary>
			/// Used to decide if the given observers return value should be included as a result
			/// if the message recieved needs to have a reply.
			/// </summary>
			public bool ProvidesResult;

			/// <summary>
			/// A pattern to match the message body against to determin if this observer should be called
			/// when a message comes in.  Can be null, which indicates that no pattern matching is being
			/// used and the observer should be called every time.  The actual pattern matching is done 
			/// through Swaf.Resource.ResourceManager.match(); see it for more information on how
			/// patterns can be formed.
			/// </summary>
			public string MatchPattern;
			/// <summary>
			/// Used to decide whether and how returned results should be encrypted or not.  If this
			/// option is set and the cryptogrophy feature is not included, an exception will occur
			/// </summary>
			public string EncryptResults;
			/// <summary>
			/// Used to determine whether or not to try and send the IMessage object to the call.
			/// </summary>
			public bool MessageStringOnly=false;

			public ObserverInfo(ICall obs, string formatter, bool providesResult, string matchPattern, string encryptResults)
			{
				Observer = obs;
				Formatter = formatter;
				ProvidesResult = providesResult;
				MatchPattern = matchPattern;
				EncryptResults = encryptResults;
			}
		}

	}

	public class MessagingListenerMessageInfo
	{
		public MessagingListener Listener;
		public IMessage Msg;
		public MessagingListenerMessageInfo(MessagingListener listener, IMessage msg)
		{
			Listener = listener;
			Msg = msg;
		}
	}

	public class MessagingListenerMessageProcessingRule
	{
		public void messagingListenerProcessMessage(IApplication app, IRuleContext context, MessagingListenerMessageInfo info)
		{
			info.Listener.doMessageProcessingWork(info.Msg);
		}
	}
}
